perf(stream): extract reportStageError out of GraphInterpreter.execute hot loop#3119
Merged
Merged
Conversation
…e hot loop Motivation: GraphInterpreter.execute is invoked once per event in the stream engine and shows up as a ~10% CPU hotspot in flamegraphs of high-throughput gRPC workloads. The nested `def reportStageError` closure inside the while-loop forced the Scala compiler to emit closure-capture bytecode (loading activeStage, chasedPush, chasedPull, chaseCounter, log, logics) on every iteration, even though every captured name is already a class field and the closure is only invoked on the error path. Modification: Lift `reportStageError` to a private[stream] instance method of GraphInterpreter, placed between `shutdownCounters` and `execute`. The three try/catch sites in execute (normal dispatch, chased PUSH, chased PULL) now call the method directly. Rename the outer `catch case NonFatal(e)` local to `ex` to remove name-shadowing with the method parameter. Result: - The hot while-loop body in `execute` shrinks by ~30 lines of bytecode, giving the JIT a tighter compile target and a better chance of inlining `processEvent`/`processPush`. - Zero per-iteration closure-capture overhead (previously the nested `def` produced a fresh closure allocation each loop pass before JIT escape analysis could eliminate it). - Behavior is preserved: `GraphInterpreterSpec` 11/11 tests pass (identity/chained/ detacher/zip/broadcast/merge/balance/cycle). - Incidental benefit: error-path logging is now declared once and easier to audit. Tests: - sbt "stream-tests/Test/testOnly org.apache.pekko.stream.impl.fusing.GraphInterpreterSpec" -> 11 passed, 0 failed. References: None - internal stream engine refactor.
pjfanning
reviewed
Jun 22, 2026
pjfanning
approved these changes
Jun 22, 2026
pjfanning
left a comment
Member
There was a problem hiding this comment.
lgtm
A good candidate for backporting to 1.7.x branch
Motivation: An independent review of apache#3119 identified three inaccuracies in the original submission: 1. The claim that the nested `def reportStageError` inside `execute` allocated a fresh closure per loop iteration was incorrect. Scala 2.13 compiles the nested `def` as a private synthetic method on the enclosing class when only class fields are referenced — verified via `javap` against the baseline bytecode (`private final void reportStageError$1(java.lang.Throwable)`). 2. The claim that `ActorGraphInterpreter` referenced the lifted method was false — a search of the file yields zero references. 3. The `private[stream]` visibility granted package-wide access that no caller actually uses. Modification: - Rewrite the scaladoc on `reportStageError` to describe the real benefit: smaller `execute` bytecode for a tighter JIT compile target on the hot `processPush`/`processPull` paths that follow each try/catch. No allocation reduction is claimed. - Narrow visibility from `private[stream]` to `private` (all three call sites live in `GraphInterpreter.execute` itself). - Rename the catch-bound local `e` to `ex` at the remaining two call sites (chased PUSH and chased PULL) so all three try/catch blocks use the same naming convention. The main-dispatch call site was already renamed in the previous commit. Result: - Semantics preserved character-for-character (same log-level dispatch, same `failStage`, same chasing-abort branch). - `GraphInterpreterSpec` 11/11 and `GraphInterpreterFailureModesSpec` 9/9 pass locally; the existing failure spec already provides directional coverage of the refactored method through every failure hook. - PR description updated in apache#3119 to reflect the corrected justification and to flag the JMH variance caveat. Tests: - sbt "stream-tests/Test/testOnly org.apache.pekko.stream.impl.fusing.GraphInterpreterSpec org.apache.pekko.stream.impl.fusing.GraphInterpreterFailureModesSpec" -> 20 passed, 0 failed. References: Refs apache#3119
Member
Author
|
I think we can have better numbers in Pekko GRPC 2.0.0. https://github.com/LesnyRumcajs/grpc_bench |
pjfanning
reviewed
Jun 22, 2026
Motivation: A maintainer nit on apache#3119 flagged the catch-bound local rename `e` -> `ex` introduced in the previous commit as unnecessary. There is no real shadowing: the catch-bound `e` and the method parameter `e` on `reportStageError` live in completely different lexical scopes, so the rename adds no clarity and diverges from the surrounding codebase convention. Modification: Revert `case NonFatal(ex) => reportStageError(ex)` back to `case NonFatal(e) => reportStageError(e)` at all three call sites in `GraphInterpreter.execute` (normal dispatch, chased PUSH, chased PULL). No other changes. Result: - Code matches the surrounding convention in the stream engine. - Semantics identical (catch-bound `e` still bound per call site, method parameter `e` still bound per invocation). - `GraphInterpreterSpec` 11/11 and `GraphInterpreterFailureModesSpec` 9/9 pass locally. - PR description updated to record the revert as reviewer-driven revision apache#5. Tests: - sbt "stream-tests/Test/testOnly org.apache.pekko.stream.impl.fusing.GraphInterpreterSpec org.apache.pekko.stream.impl.fusing.GraphInterpreterFailureModesSpec" -> 20 passed, 0 failed. References: Refs apache#3119
Motivation: A maintainer comment on apache#3119 noted that `@InternalApi` on a class-scoped `private def` is meaningless — the annotation documents that a *non-private* method is part of Pekko's internal API and may change without notice. When the method is already `private` to the enclosing class, no external caller can see it regardless, so the annotation adds no information. Modification: Remove `@InternalApi` from `reportStageError`. The `InternalApi` import is retained because it is still used on the enclosing `GraphInterpreter` companion object, the class itself, and on `activeStage` / `nonNull` which are `private[stream]` (package-visible) and therefore still warrant the annotation. Result: - No behavioral change; the method remains class-private and its semantics are preserved character-for-character. - `GraphInterpreterSpec` 11/11 and `GraphInterpreterFailureModesSpec` 9/9 pass locally. Tests: - sbt "stream-tests/Test/testOnly org.apache.pekko.stream.impl.fusing.GraphInterpreterSpec org.apache.pekko.stream.impl.fusing.GraphInterpreterFailureModesSpec" -> 20 passed, 0 failed. References: Refs apache#3119
Member
Author
|
Addressed both maintainer nits in two follow-up commits on this branch:
Both commits verified locally: PR description updated to record these as reviewer-driven revisions. Thanks for the careful review. |
Member
|
@He-Pin is this worth backporting to 1.7.x? |
Member
Author
|
I don't think so, this is just a micro optimization, wdyt, otherwise we have much to backport. |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Motivation
GraphInterpreter.executeis on the hottest path of the stream engine — it runs once per queued event. Inside the main event loop it declares a localdef reportStageError(e: Throwable), which the Scala 2.13 compiler lifts into aprivate finalsynthetic method on the enclosing class (not a per-iteration closure allocation, since the body only references class-level fields and methods). Lifting it explicitly out of the loop makes the intent obvious, shrinks the body ofexecute, and gives the JIT a tighter compile target for inlining the genuinely hotprocessPush/processPullpaths that follow each catch block.Modification
Move the nested
reportStageErrordefinition out ofexecuteinto aprivateinstance method onGraphInterpreter, placed betweenshutdownCountersandexecute. The three call sites inexecute(normal dispatch, chased PUSH, chased PULL) now invoke the instance method directly. The method body and the catch-bound locals at the call sites are unchanged.Visibility is
private(notprivate[stream]) because the only call sites live inGraphInterpreter.executeitself —ActorGraphInterpreterdoes not reference this method, and there is no planned external caller. If one materialises, widening later is a one-line change.Because
activeStage,chasedPush,chasedPull,chaseCounter,log,logics,defaultErrorReportingLogLevel,enqueue, andafterStageHasRunare all fields/methods of the enclosingGraphInterpreter, the lifted method observes exactly the same state as the previous nesteddef, and the chasing abort semantics (resetchaseCounter, re-enqueue pending chased push/pull) are preserved character-for-character.Result
executebytecode → friendlier to JIT inlining of the hotprocessPush/processPullcalls that follow each try/catch.LogLevelsattribute, callactiveStage.failStage, resetchaseCounter, and re-enqueue any pending chased push/pull.defas a synthetic method when only class fields are referenced.Benchmarks
Initial JMH data with
InterpreterBenchmark(graph_interpreter_100k_elements, 100k elements per invocation) showed improvement in the shallow-pipeline case:numberOfIdsCaveats, flagged by the reviewer:
numberOfIds=1(±7,622, ~17% relative) is far wider than the PR run (±1,222, ~2.4%), and the two confidence intervals overlap. This suggests the baseline run was noisy (GC, thermal, background processes) and the +10.6% should be treated as indicative rather than statistically decisive.The honest takeaway is: this change produces smaller, cleaner bytecode for
execute, which is genuinely beneficial for JIT compilation regardless of the exact throughput delta. Reviewers should treat any performance claim here as "likely small, not regressive, and hard to measure at this level."Tests
sbt "stream-tests/Test/testOnly org.apache.pekko.stream.impl.fusing.GraphInterpreterSpec"— 11/11 passed.sbt "stream-tests/Test/testOnly org.apache.pekko.stream.impl.fusing.GraphInterpreterFailureModesSpec"— 9/9 passed. All 9 failure scenarios (failure inonPull,onPush,onUpstreamFinish,onUpstreamFailure,onDownstreamFinish,preStart,postStop, plus pending cancel/complete) exercisereportStageErrorthroughexecute's catch blocks.sbt "bench-jmh/Jmh/run ... .*InterpreterBenchmark.*"— results above.No new test was added. The existing
GraphInterpreterFailureModesSpecalready provides directional coverage of the refactored method through every failure hook; a separate test targetingreportStageErrorin isolation would duplicate that coverage without improving regression detection (the method isprivateand reachable only throughexecute, which is precisely what the existing specs invoke).Reviewer-Driven Revisions
Following an independent review, the following were corrected from the initial submission:
defas a synthetic method on the enclosing class when only class fields are referenced, so no closure is allocated per iteration.private[stream]toprivate(the only callers are in this class;ActorGraphInterpreterdoes not reference the method).ActorGraphInterpreterusage claim from the description.executebytecode for JIT) rather than allocation reduction.etoex— per maintainer nit, there is no real shadowing since the catch-boundeand the method parameterelive in completely different scopes.References
None.